/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */



package com.bff.gaia.unified.sdk.transforms.xjoin.core.side;

/*

 * Licensed to the Apache Software Foundation (ASF) under one or more

 * contributor license agreements.  See the NOTICE file distributed with

 * this work for additional information regarding copyright ownership.

 * The ASF licenses this file to You under the Apache License, Version 2.0

 * (the "License"); you may not use this file except in compliance with

 * the License.  You may obtain a copy of the License at

 *

 *    http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */



import com.google.common.collect.Lists;

import com.google.common.collect.Maps;



import java.io.Serializable;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;



import com.bff.gaia.unified.sdk.schemas.Schema;

import com.bff.gaia.unified.sdk.transforms.xjoin.core.enums.EJoinType;

import com.bff.gaia.unified.sdk.transforms.xjoin.core.side.cache.AbsSideCache;



/**

 * Description: gaia-parent Created by tiger on 2019/7/18 16:56.

 */

public abstract class SideInfo implements Serializable {



	protected Schema rowTypeInfo;



	protected List<FieldInfo> outFieldInfoList;



	protected List<String> equalFieldList = Lists.newArrayList();



	protected List<Integer> equalValIndex = Lists.newArrayList();



	protected String sqlCondition = "";



	protected String sideSelectFields = "";



	protected EJoinType joinType;



	// key:Returns the value of the position, value: the ref field index​in the input table

	protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();



	// key:Returns the value of the position, value:  the ref field index​in the side table

	protected Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();



	// key:Returns the value of the position, value:  the ref key field index​in the side table

	protected List<Integer> outFieldIndex = new ArrayList<>();

	protected int[] outFieldIndexs;



	// key:Returns the value of the position, value:  the ref field name​in the side table

	protected Map<Integer, String> sideFieldNameIndex = Maps.newHashMap();



	protected SideTableInfo sideTableInfo;



	protected AbsSideCache sideCache;



	protected int projectNum;



	public SideInfo(

			Schema rowTypeInfo,

			JoinInfo joinInfo,

			List<FieldInfo> outFieldInfoList,

			SideTableInfo sideTableInfo) {

		this.rowTypeInfo = rowTypeInfo;

		this.outFieldInfoList = outFieldInfoList;

		this.joinType = joinInfo.getJoinType();

		this.sideTableInfo = sideTableInfo;

		projectNum = outFieldInfoList.size();

		parseSelectFields(joinInfo);

		buildEqualInfo(joinInfo, sideTableInfo);

		buildOutFieldIndex();

	}



	public void parseSelectFields(JoinInfo joinInfo) {

		String sideTableName = joinInfo.getSideTableName();

		String nonSideTableName = joinInfo.getNonSideTable();

		List<String> fields = Lists.newArrayList();



		for (int i = 0; i < outFieldInfoList.size(); i++) {

			FieldInfo fieldInfo = outFieldInfoList.get(i);

			if (fieldInfo.getTable().equalsIgnoreCase(sideTableName)) {

				if (fieldInfo.isJoinKey()) {

					dealOneEqualCon(fieldInfo.getName(), true);

					projectNum--;

				} else {

					outFieldIndex.add(i);

				}

				fields.add(fieldInfo.getName());

				sideFieldIndex.put(i, sideTableInfo.getFieldList().indexOf(fieldInfo.getName()));

				sideFieldNameIndex.put(i, fieldInfo.getName());

				//				sideIndex++;

			} else if (fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)) {

				if (fieldInfo.isJoinKey()) {

					dealOneEqualCon(fieldInfo.getName(), false);

				}

				int nonSideIndex = rowTypeInfo.indexOf(fieldInfo.getName());

				outFieldIndex.add(i);

				inFieldIndex.put(i, nonSideIndex);

			} else {

				throw new RuntimeException("unknown table " + fieldInfo.getTable());

			}

		}



		if (fields.size() == 0) {

			throw new RuntimeException("select non field from table " + sideTableName);

		}



		sideSelectFields = String.join(",", fields);

	}



	public void dealOneEqualCon(String equalFieldName, boolean isSideTable) {



		int equalFieldIndex = -1;

		if (isSideTable) {



			List<String> sideTalbeFields = sideTableInfo.getFieldList();

			for (int i = 0; i < sideTalbeFields.size(); i++) {

				if (sideTalbeFields.get(i).equals(equalFieldName)) {

					equalFieldIndex = i;

					equalFieldList.add(equalFieldName);

					break;

				}

			}

			if (equalFieldIndex == -1) {

				throw new RuntimeException("can't find equal field " + equalFieldName);

			}

		} else {

			equalFieldIndex = rowTypeInfo.indexOf(equalFieldName);

			equalValIndex.add(equalFieldIndex);

		}

	}



	public abstract void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo);



	public void buildOutFieldIndex() {

		outFieldIndexs = new int[outFieldIndex.size()];

		for (int i = 0; i < outFieldIndexs.length; i++) {

			outFieldIndexs[i] = outFieldIndex.get(i);

		}

	}



	public List<FieldInfo> getOutFieldInfoList() {

		return outFieldInfoList;

	}



	public void setOutFieldInfoList(List<FieldInfo> outFieldInfoList) {

		this.outFieldInfoList = outFieldInfoList;

	}



	public List<String> getEqualFieldList() {

		return equalFieldList;

	}



	public void setEqualFieldList(List<String> equalFieldList) {

		this.equalFieldList = equalFieldList;

	}



	public List<Integer> getEqualValIndex() {

		return equalValIndex;

	}



	public void setEqualValIndex(List<Integer> equalValIndex) {

		this.equalValIndex = equalValIndex;

	}



	public String getSqlCondition() {

		return sqlCondition;

	}



	public void setSqlCondition(String sqlCondition) {

		this.sqlCondition = sqlCondition;

	}



	public String getSideSelectFields() {

		return sideSelectFields;

	}



	public void setSideSelectFields(String sideSelectFields) {

		this.sideSelectFields = sideSelectFields;

	}



	public EJoinType getJoinType() {

		return joinType;

	}



	public void setJoinType(EJoinType joinType) {

		this.joinType = joinType;

	}



	public Map<Integer, Integer> getInFieldIndex() {

		return inFieldIndex;

	}



	public void setInFieldIndex(Map<Integer, Integer> inFieldIndex) {

		this.inFieldIndex = inFieldIndex;

	}



	public Map<Integer, Integer> getSideFieldIndex() {

		return sideFieldIndex;

	}



	public void setSideFieldIndex(Map<Integer, Integer> sideFieldIndex) {

		this.sideFieldIndex = sideFieldIndex;

	}



	public SideTableInfo getSideTableInfo() {

		return sideTableInfo;

	}



	public void setSideTableInfo(SideTableInfo sideTableInfo) {

		this.sideTableInfo = sideTableInfo;

	}



	public AbsSideCache getSideCache() {

		return sideCache;

	}



	public void setSideCache(AbsSideCache sideCache) {

		this.sideCache = sideCache;

	}



	public Map<Integer, String> getSideFieldNameIndex() {

		return sideFieldNameIndex;

	}



	public void setSideFieldNameIndex(Map<Integer, String> sideFieldNameIndex) {

		this.sideFieldNameIndex = sideFieldNameIndex;

	}



	public int getProjectNum() {

		return projectNum;

	}



	public void setProjectNum(int projectNum) {

		this.projectNum = projectNum;

	}



	public List<Integer> getOutFieldIndex() {

		return outFieldIndex;

	}



	public void setOutFieldIndex(List<Integer> outFieldIndex) {

		this.outFieldIndex = outFieldIndex;

	}



	public int[] getOutFieldIndexs() {

		return outFieldIndexs;

	}



	public void setOutFieldIndexs(int[] outFieldIndexs) {

		this.outFieldIndexs = outFieldIndexs;

	}



	public Schema getRowTypeInfo() {

		return rowTypeInfo;

	}



	public void setRowTypeInfo(Schema rowTypeInfo) {

		this.rowTypeInfo = rowTypeInfo;

	}

}